package com.xschen.kafka.demo.controller;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.xschen.kafka.demo.constants.CommonConstants;
import com.xschen.kafka.demo.dto.Payload;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Map;

/**
 * @author xschen
 */

@RestController
@RequestMapping("/api/v1")
public class KafkaProducerController {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private ObjectMapper mapper;

    /**
     * 测试
     * @param payload
     * @return
     */
    @PostMapping(value = "/sendMsg/testMsg",
            produces = MediaType.APPLICATION_JSON_VALUE)
    public Map<String, String> sendTestMsg(@RequestBody Payload payload) {
        try {
            String value = mapper.writeValueAsString(payload.getData());
            kafkaTemplate.send(CommonConstants.TEST_TOPIC, value);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }

        return payload.getData();
    }

    /**
     * 网络攻击接口
     * @param payload
     * @return
     */
    @PostMapping(value = "/sendMsg/netAttack",
            produces = MediaType.APPLICATION_JSON_VALUE)
    public String sendNetAttackMsg(@RequestBody Payload payload) {
        try {
            String value = mapper.writeValueAsString(payload.getData());
            kafkaTemplate.send(CommonConstants.NET_ATTACK, value);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        return "success";
    }

    /**
     * ddos 接口
     * @param payload
     * @return
     */
    @PostMapping(value = "/sendMsg/ddos",
            produces = MediaType.APPLICATION_JSON_VALUE)
    public String sendDdosMsg(@RequestBody Payload payload) {
        try {
            String value = mapper.writeValueAsString(payload.getData());
            kafkaTemplate.send(CommonConstants.DDOS, value);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        return "success";
    }

    /**
     * 病毒感染数据接口
     * @param payload
     * @return
     */
    @PostMapping(value = "/sendMsg/virus",
            produces = MediaType.APPLICATION_JSON_VALUE)
    public String sendVirusMsg(@RequestBody Payload payload) {
        try {
            String value = mapper.writeValueAsString(payload.getData());
            kafkaTemplate.send(CommonConstants.VIRUS, value);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        return "success";
    }

    /**
     * 防病毒产品安装数接口
     * @param payload
     * @return
     */
    @PostMapping(value = "/sendMsg/virusDevice",
            produces = MediaType.APPLICATION_JSON_VALUE)
    public String sendVirusDeviceMsg(@RequestBody Payload payload) {
        try {
            String value = mapper.writeValueAsString(payload.getData());
            kafkaTemplate.send(CommonConstants.VIRUS_DEVICE, value);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        return "success";
    }

    /**
     * 恶意邮件数据接口
     * @param payload
     * @return
     */
    @PostMapping(value = "/sendMsg/maliceMail",
            produces = MediaType.APPLICATION_JSON_VALUE)
    public String sendMaliceMailMsg(@RequestBody Payload payload) {
        try {
            String value = mapper.writeValueAsString(payload.getData());
            kafkaTemplate.send(CommonConstants.MALICE_MAIL, value);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        return "success";
    }

    /**
     * 反垃圾邮件接口
     * @param payload
     * @return
     */
    @PostMapping(value = "/sendMsg/garbageMail",
            produces = MediaType.APPLICATION_JSON_VALUE)
    public String sendGarbageMailMsg(@RequestBody Payload payload) {
        try {
            String value = mapper.writeValueAsString(payload.getData());
            kafkaTemplate.send(CommonConstants.GARBAGE_MAIL, value);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        return "success";
    }

    /**
     * 假冒网站/App 接口
     * @param payload
     * @return
     */
    @PostMapping(value = "/sendMsg/phishing",
            produces = MediaType.APPLICATION_JSON_VALUE)
    public String sendPhishingMsg(@RequestBody Payload payload) {
        try {
            String value = mapper.writeValueAsString(payload.getData());
            kafkaTemplate.send(CommonConstants.PHISHING, value);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        return "success";
    }

    /**
     * 黑设备指纹
     * @param payload
     * @return
     */
    @PostMapping(value = "/sendMsg/devFp",
            produces = MediaType.APPLICATION_JSON_VALUE)
    public String sendDevFpMsg(@RequestBody Payload payload) {
        try {
            String value = mapper.writeValueAsString(payload.getData());
            kafkaTemplate.send(CommonConstants.DEV_FP, value);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        return "success";
    }

    /**
     * 黑设备封禁
     * @param payload
     * @return
     */
    @PostMapping(value = "/sendMsg/devBlock",
            produces = MediaType.APPLICATION_JSON_VALUE)
    public String sendDevBlockMsg(@RequestBody Payload payload) {
        try {
            String value = mapper.writeValueAsString(payload.getData());
            kafkaTemplate.send(CommonConstants.DEV_BLOCK, value);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        return "success";
    }

    /**
     * 信息科技外包评价数据
     * @param payload
     * @return
     */
    @PostMapping(value = "/sendMsg/outsource",
            produces = MediaType.APPLICATION_JSON_VALUE)
    public String sendOutSourceMsg(@RequestBody Payload payload) {
        try {
            String value = mapper.writeValueAsString(payload.getData());
            kafkaTemplate.send(CommonConstants.OUTSOURCE, value);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        return "success";
    }
}
